package com.alan.weather.service.weather;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

/**
 * HadoopUtils封装类
 * 封装hadoop常用操作
 * 配置对象
 * 获取文件系统
 * 删除目录
 * 显示指定目录下文件内容
 *
 * @author LX
 */
public abstract class HadoopUtils {
    private static Configuration conf;

    static {
        // 实例化配置对象
        conf = new Configuration();
        // 设置hadoop集群属性，若提供xml文件，自动读取，无需设置
        // conf.set("fs.defaultFS", "hdfs://master:9000");
    }

    /**
     * 返回HDFS文件系统对象
     *
     * @return HDFS文件系统对象
     */
    public static FileSystem getFileSystem() throws IOException {
        return FileSystem.get(conf);
    }

    public static void delete(FileSystem hdfs, Path dst, boolean isDeleted) throws IOException {
        if (hdfs.exists(dst)){
            hdfs.delete(dst, isDeleted);
        }else {
            return;
        }
    }

    /**
     * 显示指定目录下所有内容
     */
    public static void showContent(FileSystem hdfs, Path dst) throws IOException {
        for (FileStatus f : hdfs.listStatus(dst)) {
            // 如果是文件，则读取内容
            if (f.isFile()) {
                // 构建输入流
                FSDataInputStream inputStream = hdfs.open(f.getPath());
                // 构建缓存读取器
                BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
                // 循环按行读取内容
                String line = reader.readLine();
                while (line != null) {
                    // 输出内容
                    System.out.println(line);
                    line = reader.readLine();
                }
                // 关闭读取器
                reader.close();
                // 关闭流
                inputStream.close();
            }
        }
    }
    /**
     * 返回Configuration对象属性
     *
     * @return Configuration对象属性
     */
    public static Configuration getConf() {
        return conf;
    }
}
